package com.hqyj.bigdata_project.util;

import com.hqyj.bigdata_project.mapper.ETLMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.File;

public class ETLDriver implements Tool {

    private Configuration configuration;

    public int run(String[] args) throws Exception {
        /*
        args = new String[]{"F:\\myvideo\\2008\\0222","f:/myvideo/out"};
        */

        args = new String[]{"F:"+ File.separator+"myvideo"+File.separator+"2008"+File.separator+"0222","F:"+File.separator+"myvideo"+File.separator+"out"};

        //1.获取Job
        Job job = Job.getInstance(configuration);

        //2.设置Jar包路径
        job.setJarByClass(ETLDriver.class);

        //3.设置Mapper类输出Kv类型
        job.setMapperClass(ETLMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        //job.setOutputValueClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //4.设置最终输出的KV类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        //5.设置输入输出的路径
        FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        //6.提交任务
        boolean result = job.waitForCompletion(true);

        return result?0:1;

    }

    public void setConf(Configuration conf) {
        configuration = conf;
    }

    public Configuration getConf() {
        return configuration;
    }


    public static void main(String[] args) {
        try {
            //构建配置信息
            Configuration configuration = new Configuration();
            int run = ToolRunner.run(configuration, new ETLDriver(), args);
            System.out.println(run);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
